 
package com.dabarobjects.mobile.database;

import com.dabarobjects.mobile.database.types.BooleanObjectNotationParser;
import com.dabarobjects.mobile.database.types.ComplexObjectNotationParser;
import com.dabarobjects.mobile.database.types.DateObjectNotationParser;
import com.dabarobjects.mobile.database.types.IntegerObjectNotationParser;
import com.dabarobjects.mobile.database.types.StringObjectNotationParser;
import com.dabarobjects.mobile.utils.DataUtils;
import com.dabarobjects.mobile.utils.logger.EventLogger;
import com.dabarobjects.mobile.utils.logger.LogManager;
import com.dabarobjects.mobile.utils.madex.MadexEncoder;
import com.dabarobjects.mobile.utils.madex.MadexParserUtils;
import java.util.Date;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Random;
import java.util.Vector;
import javax.microedition.lcdui.Command;
import javax.microedition.lcdui.CommandListener;
import javax.microedition.lcdui.Displayable;
import javax.microedition.rms.RecordEnumeration;
import javax.microedition.rms.RecordStore;
import javax.microedition.rms.RecordStoreException;
import javax.microedition.rms.RecordStoreNotOpenException;

/**
 *
 * A Database Management System for Mobile Phones. A simple Key-Value Implementation allowing users to
 * put Objects into the data store using unique keys and retrieve them later with the key
 * 
 * @author (Deji Aladejebi) DabarObjects Solutions Limited
 * @version 1.00
 * @since 2010
 */
public class MadexDBMS implements CommandListener {

    private RecordStore rs;
    private EventLogger logger;

    public MadexDBMS() {
        logger = LogManager.getInfoLogger();
        started = false;
    }

    public MadexDBMS(EventLogger log) {
        logger = log;
        started = false;
    }
    private volatile boolean started;

    public synchronized boolean startDB(String dababaseName) {
        if (!started) {
            started = openStore(dababaseName);

            //create index
            createIndex();
            //REGISTER TYPE PARSERS
            DBMSObjectParser.get().registerType(new DateObjectNotationParser());
            DBMSObjectParser.get().registerType(new StringObjectNotationParser());
            DBMSObjectParser.get().registerType(new BooleanObjectNotationParser());
            DBMSObjectParser.get().registerType(new IntegerObjectNotationParser());
            DBMSObjectParser.get().registerType(new ComplexObjectNotationParser());

        }

        return started;
    }
    public void turnOffLogging(){
        logger.turnOffAndClear();
    }
    public void turnOnLogging(){
        logger.turnOn();
    }
    /**
     * Indexes are made in the following form;
     * KeyName1-RecId1;KeyName2-RecId2
     */
    private final static String INDEX_KEY = "INDEX";

    private void createIndex() {
        try {
            indexLocation = 1;
            ////System.out.println("Index Location: " + indexLocation);
            if (rs.getNumRecords() > 0) {
                //indexRecord(INDEX_KEY, indexLocation);
            } else {
                //create new index
                createNewIndex(INDEX_KEY, indexLocation);
            }
        } catch (RecordStoreNotOpenException recordStoreNotOpenException) {
            logger.error(recordStoreNotOpenException);
        }

    }

    /**
     * Adds a RecordItem into the data store. The Method overloads addRecord by taking in a key and a value parameters
     * @param key
     * @param value
     */
    public void saveRecord(String key, String value) {
        save(key, value);
    }

    public void updateRecord(String key, String value) {

        int recId = findIndex(key);
        if (recId != -1) {
            update(key, value);
        } else {
            saveRecord(key, value);
        }

    }

    /**
     * Deletes a RecordItem from the data store. Returns Zero if successful or -1 if not
     * @param key
     */
    public int deleteRecord(String key) {
        if (key.equalsIgnoreCase(INDEX_KEY)) {
            if (logger.isDebug()) {
                logger.info("deleteRecord(" + key + ")", "Data Key Cannot Be Named: " + key);
            }
            return -1;
        }
        try {

            int recId = findIndex(key);
            if (recId != -1) {
                rs.deleteRecord(recId);
                deleteIndex(key);
                return 0;
            }

            return -1;
        } catch (RecordStoreException recordStoreNotOpenException) {
            logger.error(recordStoreNotOpenException);
            return -1;
        }
    }
    private int indexLocation;

    /**
     * Indexes an entry into the database. The index works by keeping the recordID of an item against the Key of the item. This way, each time
     * we need an item, we can simply search the index to locate the recordId of the item. Indexes data are kept in the first record of the store
     * it is initialized at the begining of the application
     * Indexes are store as KeyName1-RecId1;KeyName2-RecId2; pattern
     * Index data is stored in record 0
     */
    private void indexRecord(String keyName, int recordId) {
        try {
            Hashtable indexTable = loadIndexes();

            indexTable.put(keyName, new Integer(recordId));
            //index has become stale so we update
            updateIndexStore(indexTable);
        } catch (RecordStoreException ex) {
            logger.error(ex);

        }

    }

    /**
     * Loads the index and returns them as a Hashtable for an Add Record Operation to create new indexes. The Hash table upon modification must be written back to
     * string and persisted to become valid
     * @return
     * @throws RecordStoreException
     */
    private synchronized Hashtable loadIndexes() throws RecordStoreException {
        byte[] indexData = rs.getRecord(indexLocation);
        String indexDataStr = new String(indexData);
        return stringToHashTable(indexDataStr, "-");

    }

    /**
     * For a Multi row record set i.e. a key/value pair with mutiple rows, indexes of new records are store by their parent key position so that
     * a getList can return every related record easily
     * Direct Indexes are stored as KeyName1-RecId1#RecId2#RecId3; pattern
     * so that a single call to KeyName1 will return all items in 1,2,3
     * @param keyName
     * @param recordId
     */
    private void indexRecordDirect(String keyName, int recordId) {
        try {
            //get Index
            Hashtable indexTable = loadIndexes();

            //Do what you have to do with index
            String existData = (String) indexTable.get(keyName);

            if (existData != null) {
                existData = new StringBuffer().append(existData).append("#").append(recordId).toString();
            } else {
                existData = new StringBuffer().append(recordId).toString();
            }
            indexTable.put(keyName, existData);

            logger.info("indexRecordDirect(" + keyName + "," + recordId + ")", indexTable.toString());
            //Write index back
            updateIndexStore(indexTable);
        } catch (RecordStoreException ex) {
            logger.error(ex);

        }

    }

    /**
     * Done at the creation of a new database
     * @param keyName
     * @param recordId
     */
    private void createNewIndex(String keyName, int recordId) {
        try {
            final Hashtable indexTable = new Hashtable();
            indexTable.put(keyName, new Integer(recordId));
            String newIndexDataStr = hashTableToString(indexTable, "-");
            byte[] newIndexDataByte = newIndexDataStr.getBytes();
            indexLocation = rs.addRecord(newIndexDataByte, 0, newIndexDataByte.length);
            logger.info("createNewIndex(" + keyName + ")", "Index  :" + indexLocation);

        } catch (RecordStoreException recordStoreException) {
            logger.error(recordStoreException);
        }

    }

    /**
     * Tracks the next index
     * NOT USED
     * @deprecated 
     * @return
     */
    private int trackIndex() {
        try {
            RecordEnumeration recEnum = rs.enumerateRecords(null, null, true);
            if (recEnum.numRecords() > 0) {
                return recEnum.nextRecordId();
            }
            return rs.getNextRecordID();
        } catch (Exception recordStoreNotOpenException) {
            recordStoreNotOpenException.printStackTrace();
            return 0;
        }
    }

    /**
     * Removes an index from the index store
     * @param keyName
     */
    private void deleteIndex(String keyName) {
        if (keyName.equalsIgnoreCase(INDEX_KEY)) {
            logger.info("deleteIndex(" + keyName + ")", "Data Key Cannot Be Named: " + keyName);
            return;
        }
        try {
            byte[] indexData = rs.getRecord(indexLocation);
            String indexDataStr = new String(indexData);
            Hashtable indexTable = stringToHashTable(indexDataStr, "-");
            indexTable.remove(keyName);
            updateIndexStore(indexTable);
        } catch (RecordStoreException ex) {
            logger.error(ex);
        }

    }

    /**
     * Writes the modifed index back to storage. Indexing operations are not yet fully atomic
     * @param indexTable
     */
    private synchronized void updateIndexStore(Hashtable indexTable) {
        try {
            String newIndexDataStr = hashTableToString(indexTable, "-");
            logger.info("updateIndexStore()", newIndexDataStr);
            byte[] newIndexDataByte = newIndexDataStr.getBytes();
            rs.setRecord(indexLocation, newIndexDataByte, 0, newIndexDataByte.length);

        } catch (RecordStoreException ex) {
            logger.error(ex);
        }

    }

    public RecordItem findRecord2(String keyName) {
        try {

            int recId = findIndex(keyName);
            if (recId != -1) {
                byte[] recordByteSet = rs.getRecord(recId);
                return translateNotation(recordByteSet);
            }

        } catch (RecordStoreException recordStoreNotOpenException) {
            logger.error(recordStoreNotOpenException);
        }
        return null;
    }

    /**
     * Very critical method is used to track index for any Key. called to return the index for a single Key/Pair combination
     * @param keyName
     * @return the index if found and -1 if not
     */
    private int findIndex(String keyName) {
        try {
            //Load indexes
            Hashtable indexTable = loadIndexes();

            if (indexTable.containsKey(keyName)) {
                int recId = Integer.parseInt((String) indexTable.get(keyName));
                return recId;
            }

            if (logger.isDebug()) {
                logger.info("findIndex(" + keyName + ")", "Index Not Found");
            }
        } catch (RecordStoreException ex) {
            logger.error(ex);
        } finally {
        }

        return -1;
    }

    public int countRecords() {
        int recCount = 0;
        try {
            //Load indexes
            Hashtable indexTable = loadIndexes();
            recCount = indexTable.size() - 1;
            if (logger.isDebug()) {
                logger.info("findIndex()", "Index Count " + recCount);
            }
            return recCount;
        } catch (RecordStoreException ex) {
            logger.error(ex);
        }

        return -1;
    }

    /**
     * Very critical method is used to track indexes for a multi row key/pair entry
     * @param keyName
     * @return an arrray of indexes
     */
    private int[] findIndexes(String keyName) {
        try {
            //Load indexes
            Hashtable indexTable = loadIndexes();

            if (indexTable.containsKey(keyName)) {
                String indexParts = (String) indexTable.get(keyName);

                int indexes[] = null;
                int partCount = MadexParserUtils.countParts(indexParts, '#');
                indexes = new int[partCount];

                String[] partsArr = MadexParserUtils.splitUp(indexParts, '#', partCount);
                for (int i = 0; i < partsArr.length; i++) {
                    indexes[i] = Integer.parseInt(partsArr[i]);
                }
                return indexes;

            }
        } catch (RecordStoreException ex) {
            logger.error(ex);
        }

        logger.info("findIndex(" + keyName + ")", "Index Not Found");

        return null;
    }

    /**
     * Called from debuggers to inspect in case of troubleshooting. MADEX Client can simple disable logging to hide all details
     * @return
     */
    public String getIndexDump() {
        try {
            StringBuffer buf = new StringBuffer();
            //Load indexes
            Hashtable ht = loadIndexes();
            buf.append("\n");
            Enumeration em = ht.keys();
            while (em.hasMoreElements()) {
                Object indexkey = em.nextElement();
                Object indexValue = ht.get(indexkey);
                buf.append(indexkey).append(":").append(indexValue).append("\n");
                if (indexkey.toString().equalsIgnoreCase(INDEX_KEY)) {
                    continue;
                }
                int rec = Integer.parseInt(indexValue.toString());

                byte[] recordByteSet = rs.getRecord(rec);
                buf.append(translateNotation(recordByteSet).toString()).append("\n");
            }
            return buf.toString();
        } catch (Exception recordStoreException) {
            logger.error(recordStoreException);
        }
        return "Oops!";

    }

    /**
     * Called to find a record item stored against a key
     * Does not work for multi rows
     * @param keyName
     * @return returns a RecordItem or null if not found
     */
    public RecordItem findRecord(String keyName) {
        try {

            //Load indexes
            Hashtable ht = loadIndexes();

            Enumeration em = ht.keys();
            while (em.hasMoreElements()) {
                Object indexkey = em.nextElement();
                Object indexValue = ht.get(indexkey);

                if (indexkey.toString().equalsIgnoreCase(INDEX_KEY)) {
                    continue;
                }
                if (indexkey.toString().equalsIgnoreCase(keyName)) {
                    int rec = Integer.parseInt(indexValue.toString());

                    byte[] recordByteSet = rs.getRecord(rec);
                    RecordItem item = translateNotation(recordByteSet);

                    //Debug
                    if (logger.isDebug()) {
                        logger.info("findRecord(" + keyName
                                + ")", "" + item);
                    }

                    return item;
                }

            }

            if (logger.isDebug()) {
                logger.info("findRecord(" + keyName
                        + ")", keyName + ": Not Found");
            }

        } catch (Exception recordStoreException) {
            logger.error("findRecord(" + keyName
                    + ")", recordStoreException);
        }
        return null;

    }

    /**
     * Called when looking for a single row Key/Pair.
     * Does not work for multi rows
     * @param keyName
     * @return
     */
    public Object findValue(String keyName) {
        try {

            //Load indexes
            Hashtable ht = loadIndexes();

            if (logger.isDebug()) {
                logger.info("findValue(" + keyName
                        + ")", "" + ht.get(keyName));
            }
            Enumeration em = ht.keys();
            while (em.hasMoreElements()) {
                Object indexkey = em.nextElement();


                if (indexkey.toString().equalsIgnoreCase(INDEX_KEY)) {
                    continue;
                }
                if (indexkey.toString().equalsIgnoreCase(keyName)) {
                    Object indexValue = ht.get(indexkey);
                    int rec = Integer.parseInt(indexValue.toString());

                    byte[] recordByteSet = rs.getRecord(rec);

                    Object item = translateNotationDirect(recordByteSet);
                    if (logger.isDebug()) {
                        logger.info("findValue(" + keyName
                                + ")", "" + item);
                    }
                    return item;
                }

            }
            if (logger.isDebug()) {
                logger.info("findValue(" + keyName + ")", keyName + ": Not Found");
            }

        } catch (Exception recordStoreException) {
            logger.error("findRecord(" + keyName
                    + ")", recordStoreException);
        }
        return null;

    }

    /**
     * Renders the Value of the data as a viewable string
     * @param keyName
     * @return a string rep of object returned
     */
    public String findValueAsString(String keyName) {
        Object data = findValue(keyName);
        if (data instanceof String) {
            if (data != null) {
                return data.toString();
            }
        }
        if (data instanceof Date) {
            if (data != null) {
                return DataUtils.formatDateForPreview((Date) data);
            }
        }
        return null;
    }

    public Hashtable loadAllRecords() {
        Hashtable v = new Hashtable();
        try {

            //Load indexes
            Hashtable ht = loadIndexes();

            logger.info("listAll", ht.toString());
            Enumeration em = ht.keys();
            while (em.hasMoreElements()) {
                Object indexkey = em.nextElement();
                if (indexkey.toString().equalsIgnoreCase(INDEX_KEY)) {
                    continue;
                }
                Object indexValue = ht.get(indexkey);
                int rec = Integer.parseInt(indexValue.toString());
                byte[] recordByteSet = rs.getRecord(rec);

                Object data = translateNotationDirect(recordByteSet);
                v.put(indexkey, data);
            }


        } catch (Exception recordStoreException) {
            logger.error("loadAllRecords(" + ""
                    + ")", recordStoreException);
        }
        return v;
    }
    /*
     * Search across the database items and returns a vector of keys
     */

    public Vector searchRecords(String searchKey) {
        Vector v = new Vector();
        try {

            //Load indexes
            Hashtable ht = loadIndexes();

            if (logger.isDebug()) {
                logger.info("search(" + searchKey
                        + ")", "");
            }
            Enumeration em = ht.keys();
            while (em.hasMoreElements()) {
                Object indexkey = em.nextElement();
                if (indexkey.toString().equalsIgnoreCase(INDEX_KEY)) {
                    continue;
                }
                Object indexValue = ht.get(indexkey);
                int rec = Integer.parseInt(indexValue.toString());
                byte[] recordByteSet = rs.getRecord(rec);

                Object data = translateNotationDirect(recordByteSet);
                if (data instanceof String) {
                    if (data != null) {
                        String dataStr = data.toString();
                        int wordLength = dataStr.length();
                        //parse data based on searchkey length
                        int searchKeyLen = searchKey.length();
                        for (int i = 0; i < wordLength; i++) {
                            String partStr = new String(dataStr.substring(i, i + searchKeyLen));
                            if (partStr.toLowerCase().equalsIgnoreCase(searchKey.toLowerCase())) {
                                v.addElement(indexkey);
                            }
                        }
                    }
                }
            }


        } catch (Exception recordStoreException) {
            logger.error("search(" + searchKey
                    + ")", recordStoreException);
        }
        return v;
    }

    /**
     * Useful when its time to transport Madex data which may need to be encoded in case of mixes with madex representation characters
     * @param keyName
     * @return
     */
    public String findValueForTransport(String keyName) {
        Object data = findValue(keyName);
        if (data instanceof String) {
            if (data != null) {
                return MadexEncoder.encode(data.toString());
            }
        }
        if (data instanceof Date) {
            if (data != null) {
                return MadexEncoder.encode(((Date) data).toString());
            }
        }
        return "" + data;

    }

    public Date findValueAsDate(String keyName) {
        Object data = findValue(keyName);
        if (data != null) {
            return (Date) data;
        }
        return new Date();

    }

    public static String createRandomNumber(int len) {
        Random rand = new Random();
        StringBuffer id = new StringBuffer();
        for (int i = 0; i <= len; i++) {
            id.append(rand.nextInt(10));
        }
        return id.toString();
    }

    /**
     * same as findRecord(key);
     * @param key
     * @return
     */
    public RecordItem locateRecordItem(String key) {
        return findRecord(key);

    }

    private String hashTableToString(Hashtable indexMapping, String keyValueSeparator) {
        Enumeration ennKeys = indexMapping.keys();
        StringBuffer strPattBuffer = new StringBuffer();
        while (ennKeys.hasMoreElements()) {
            String keyElem = (String) ennKeys.nextElement();
            strPattBuffer.append(keyElem).append(keyValueSeparator).append(indexMapping.get(keyElem)).append(";");
        }
        return strPattBuffer.toString();
    }

    private Hashtable stringToHashTable(String indexStrData, String keyValueSeparator) {

        int len = indexStrData.length();

        Hashtable keyValue = new Hashtable();
        boolean startValueAppend = false;
        boolean startKeyAppend = true;

        /**
         * Creates the separate temp buffers for holding temporary key and value parts
         */
        StringBuffer indexValueBuffer = new StringBuffer();
        StringBuffer indexKeyBuffer = new StringBuffer();
        int pointer = 0;
        //we transverse the string and watches out for seperator patterns
        for (int i = 0; i < len; i++) {
            String it = indexStrData.substring(i, i + 1);

            if (it.equalsIgnoreCase(keyValueSeparator)) {
                pointer++;
                indexValueBuffer = new StringBuffer();
                startValueAppend = true;
                startKeyAppend = false;

            }
            if (it.equalsIgnoreCase(";")) {

                startValueAppend = false;
                startKeyAppend = true;
                String protocolHeaderKey = indexKeyBuffer.toString();
                String protocolHeaderValue = indexValueBuffer.toString();

                if (keyValue.containsKey(protocolHeaderKey)) {
                    protocolHeaderValue = protocolHeaderValue.concat("#").concat(keyValue.get(protocolHeaderKey).toString());
                }
                keyValue.put(protocolHeaderKey, protocolHeaderValue);

                indexKeyBuffer = new StringBuffer();
            }
            if (startValueAppend) {
                if (!it.equalsIgnoreCase("-")) {
                    indexValueBuffer.append(it);
                }
            }
            if (startKeyAppend) {

                if (!it.equalsIgnoreCase(";")) {
                    indexKeyBuffer.append(it);
                }
            }
        }
        return keyValue;
    }

    /**
     * Returns the database key size
     * @return
     */
    public int count() {

        try {
            //minus the index
            return rs.getNumRecords() - 1;

        } catch (RecordStoreException recordStoreNotOpenException) {
            logger.error(recordStoreNotOpenException);
        }
        return 0;

    }

    public int getMemSize() {
        try {
            int mem = rs.getSizeAvailable();
            return mem;
        } catch (RecordStoreNotOpenException recordStoreNotOpenException) {
            logger.error(recordStoreNotOpenException);
        }
        return 0;
    }

    public int getUsed() {
        try {
            int used = rs.getSize();
            return used;
        } catch (RecordStoreNotOpenException recordStoreNotOpenException) {
            logger.error(recordStoreNotOpenException);
        }
        return 0;
    }

    private boolean openStore(String dababaseName) throws RuntimeException {
        try {
            rs = RecordStore.openRecordStore(dababaseName, true);
            if (logger.isDebug()) {
                logger.info("openApp()", "Database Started");
                logger.info("openStore()", "Status: " + rs.getName() + "\nUsed Size: " + (getUsed() / 1024) + "KB" + "\nAvailable Size: " + (getMemSize() / 1024) + "KB");
                //info = false;
            }
            return true;
        } catch (RecordStoreException ex) {
            logger.error(ex);

        }
        return false;
    }

    public void shutdown() {
        try {
            rs.closeRecordStore();
            logger.clearLogData();

        } catch (RecordStoreException ex) {
            logger.error(ex);
        }
    }

    public void commandAction(Command c, Displayable d) {
        if (c.getLabel().equalsIgnoreCase("exitAppCommand")) {
        }
    }

    /**
     * called to convert an Object datatype into MADEX DB Notation
     * @param seq
     * @param key
     * @param data
     * @param update
     * @return
     * @throws UnsupportedObjectTypeOperation
     */
    private String getRecordNotation(int seq, String key, Object data, boolean update) throws UnsupportedObjectTypeOperation {
        Hashtable notationMap = new Hashtable();

        Date timestamp = new Date();

        String typeKey = "";
        if (data instanceof String) {
            typeKey = "st";
        }
        if (data instanceof Date) {
            typeKey = "dt";
        }
        if (data instanceof Integer) {
            typeKey = "nb";
        }
        if (data instanceof Boolean) {
            typeKey = "bl";
        }


        if (typeKey.equals("")) {
            throw new UnsupportedObjectTypeOperation(data);
        }

        String timeStampNotation = MadexParserUtils.dateAsString(timestamp);
        String objectNotation = DBMSObjectParser.get().notate(typeKey, data);


        notationMap.put("RK", key);
        notationMap.put("RS", String.valueOf(seq));
        notationMap.put("RV", objectNotation);
        notationMap.put("RT", typeKey);
        notationMap.put("TS", timeStampNotation);

        //String dataDump = "RK:AA;RS:3;RV:{DataType Format Pattern};RT:{Date, String, Integer};TS:{timestamp}";
        String dataDump = hashTableToString(notationMap, ":");
        ////System.out.println(dataDump);
        notationMap.clear();
        notationMap = null;
        return dataDump;
    }

    private Object translateNotationDirect(byte[] notationByte) {
        Hashtable notationMap = DBMSObjectParser.get().parse(new String(notationByte));
        String typeKey = (String) notationMap.get("RT");
        String objNotation = (String) notationMap.get("RV");
        Object data = DBMSObjectParser.get().parseObject(typeKey, objNotation);
        return data;
    }

    /**
     * Converts a MADEX DB storage notation into a useful record item object
     * @param notationByte
     * @return RecordItem which is an immutable object that encapsules the data retrieved
     */
    private RecordItem translateNotation(byte[] notationByte) {
        Hashtable notationMap = DBMSObjectParser.get().parse(new String(notationByte));

        ////System.out.println(notationMap);
        String typeKey = (String) notationMap.get("RT");
        String objNotation = (String) notationMap.get("RV");
        Object data = DBMSObjectParser.get().parseObject(typeKey, objNotation);
        int sequence = Integer.parseInt((String) notationMap.get("RS"));
        String recordKey = (String) notationMap.get("RK");
        Date timestamp = (Date) DBMSObjectParser.get().parseObject(DateObjectNotationParser.KEY, (String) notationMap.get("TS"));


        RecordItem recordObject = new RecordItem(recordKey, timestamp, data, sequence);

        return recordObject;


    }

    public synchronized void addRow(String key, Object data) {
        if (key.equalsIgnoreCase(INDEX_KEY)) {
            if (logger.isDebug()) {
                logger.info("addRow(" + key + "," + data + ")", "Data Key Cannot Be Named: " + key);
            }
            return;
        }

        try {
            //Lock the Index
            int seq = rs.getNextRecordID();

            String dataDump = getRecordNotation(seq, key, data, false);
            //encode

            logger.info("addRow(" + key + "," + data + ")", dataDump);

            byte[] recBytes = dataDump.getBytes();
            int recId = rs.addRecord(recBytes, 0, recBytes.length);
            indexRecordDirect(key, recId);
            //Unlock the index

        } catch (Exception recordStoreException) {
            logger.error(recordStoreException);
        }
    }

    /**
     * Starts a new DataSeries
     * A DataSeries allows you to further nest more key-value systems under a particular parent key
     * series are started using their own parent keys. They are more like super-columns in noSQL concept
     * its a key-value where the value holds more key-values
     * @param seriesName
     * @param colNums
     * @return 
     */
    public synchronized DataSeries startSeries(String seriesName, int colNums) {
        if (isSeriesAvailable(seriesName)) {
            return loadSeries(seriesName);
        }
        return new DataSeries(colNums, this, seriesName);
    }

    /**
     * loads an existing series or starts a new one if it does not exist
     * @param seriesKey
     * @return 
     * @see DataSeries
     */
    public DataSeries loadSeries(String seriesKey) {
        Object data = findValue(seriesKey);
        int indexSeq = findIndex(seriesKey);
        if (data == null) {
            return startSeries(seriesKey, 10);
        }

        logger.info("loadSeries(" + seriesKey + ")", data.toString());
        DataSeries itemMap = (DataSeries) MadexParserUtils
                .stringToHashTable(
                new DataSeries(this, seriesKey, indexSeq), data.toString(), "{", "}", true, false);
        return itemMap;

    }

    public boolean isSeriesAvailable(String seriesKey) {
        Object data = findValue(seriesKey);
        return data != null;
    }

    public void updateRow(String key, Object data, int seqId) {
        if (key.equalsIgnoreCase(INDEX_KEY)) {
            if (logger.isDebug()) {
                logger.info("save(" + key + "," + data + ")", "Data Key Cannot Be Named: " + key);
            }
            return;
        }
        try {
            String dataDump = getRecordNotation(seqId, key, data, true);
            byte[] recBytes = dataDump.getBytes();
            rs.setRecord(seqId, recBytes, 0, recBytes.length);


        } catch (Exception recordStoreException) {
            logger.error(recordStoreException);
        }
    }

    public int countSeries(String keyName) {
        int[] seq = findIndexes(keyName);
        if (seq == null) {
            return 0;
        }
        return seq.length;
    }

    /**
     * Creates a subset array out of a superset array
     * @param supArr
     * @param start
     * @param max
     * @return
     */
    private int[] subArray(int[] supArr, int start, int max) {
        if (start < 0) {
            return null;
        } else if (start >= supArr.length) {
            return null;
        }

        int limit = max + start;
        int size = 0;
        if (limit >= supArr.length) {
            limit = supArr.length;
        }
        size = limit - start;
        int[] nArr = new int[size];
        int k = 0;

        for (int i = start; i < limit; i++) {
            int j = supArr[i];
            nArr[k] = j;
            k++;
        }
        return nArr;
    }

    private RowItem[] loadSegment(String keyName, int[] seq) {
        RowItem[] items = null;
        try {

            if (seq == null) {
                logger.info("loadRows(null)", "Null Rows");
                return null;
            }

            items = new RowItem[seq.length];
            for (int i = 0; i < seq.length; i++) {
                int j = seq[i];
                byte[] recordByteSet = rs.getRecord(j);

                Object data = translateNotationDirect(recordByteSet);
                String typeKey = "";
                if (data instanceof String) {
                    typeKey = "st";
                }
                if (data instanceof Date) {
                    typeKey = "dt";
                }
                if (data instanceof Integer) {
                    typeKey = "nb";
                }
                RowItem rowItem = new RowItem(data, typeKey, j, keyName);
                items[i] = rowItem;

                logger.info("loadRows(" + seq.length
                        + ")", rowItem.getData().toString());


            }
        } catch (RecordStoreException recordStoreException) {
            logger.error(recordStoreException);
        }

        return items;
    }

    public RowItem[] loadRowSegment(String keyName, int start, int max) {
        int[] seq = subArray(findIndexes(keyName), start, max);
        return loadSegment(keyName, seq);
    }

    public RowItem[] loadRows(String keyName) {
        int[] seq = findIndexes(keyName);
        return loadSegment(keyName, seq);
    }
    //from 1,2

    public int deleteRow(String keyName, int rowNo) {

        return -1;
    }

    public RowItem loadRow(String keyName, int rowNo) {

        RowItem items = null;
        try {
            int[] seq = findIndexes(keyName);
            if (rowNo == 0) {
                return null;
            }
            if ((rowNo - 1) >= seq.length) {
                return null;
            }
            int seqId = seq[rowNo];
            byte[] recordByteSet = rs.getRecord(seqId);

            Object data = translateNotationDirect(recordByteSet);
            String typeKey = "";
            if (data instanceof String) {
                typeKey = "st";
            }
            if (data instanceof Date) {
                typeKey = "dt";
            }
            if (data instanceof Integer) {
                typeKey = "nb";
            }
            items = new RowItem(data, typeKey, seqId, keyName);
            return items;

        } catch (RecordStoreException recordStoreException) {
            logger.error(recordStoreException);
        }

        return items;
    }

    /**
     * Saves a new record using its key. In a typical madex enviroment, if we represent Name of Person as AA
     * We will call save("AA","dele making"); and a call to findValue("AA"); will return "dele making"
     * @param key
     * @param data
     */
    public boolean save(String key, Object data) {
        if (key.equalsIgnoreCase(INDEX_KEY)) {
            logger.info("save(" + key + "," + data + ")", "Data Key Cannot Be Named: " + key);
            throw new IllegalStateException("Data Key Cannot Be Named: " + key);
        }
        if (key == null) {
            logger.info("save(" + key + "," + data + ")", "Data Key Cannot Be Null: " + key);
            return false;
        }
        try {
            //checks if the index is located
            int seq = findIndex(key);
            if (seq != -1) {
                //rather update item if already stored
                return update(key, data);

            }

            seq = rs.getNextRecordID();

            String dataDump = getRecordNotation(seq, key, data, false);
            //encode

            logger.info("save(" + key + "," + data + ")", dataDump);

            byte[] recBytes = dataDump.getBytes();
            int recId = rs.addRecord(recBytes, 0, recBytes.length);
            indexRecord(key, recId);
            return true;
        } catch (Exception recordStoreException) {
            logger.error(recordStoreException);
        }
        return false;

    }

    /**
     * Updates a record using its key
     * @param key
     * @param data
     */
    public boolean update(String key, Object data) {
        if (key.equalsIgnoreCase(INDEX_KEY)) {
            if (logger.isDebug()) {
                logger.info("save(" + key + "," + data + ")", "Data Key Cannot Be Named: " + key);
            }
            return false;
        }
        try {
            int seq = findIndex(key);

            String dataDump = getRecordNotation(seq, key, data, true);
            byte[] recBytes = dataDump.getBytes();
            rs.setRecord(seq, recBytes, 0, recBytes.length);

            return true;
        } catch (Exception recordStoreException) {
            logger.error(recordStoreException);
        }
        return false;

    }
}
